// Copyright (C) Kumo inc. and its affiliates.
// Author: Jeff.li lijippy@163.com
// All rights reserved.
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.
//


#pragma once

#include <pollux/dwio/common/selective_column_reader_internal.h>

namespace kumo::pollux::dwio::common {
    /// Abstract class for format and encoding-independent parts of reading integer
    /// columns.
    class SelectiveIntegerColumnReader : public SelectiveColumnReader {
    public:
        SelectiveIntegerColumnReader(
            const TypePtr &requestedType,
            dwio::common::FormatParams &params,
            pollux::common::ScanSpec &scanSpec,
            std::shared_ptr<const dwio::common::TypeWithId> fileType)
            : SelectiveColumnReader(
                requestedType,
                std::move(fileType),
                params,
                scanSpec) {
        }

        void getValues(const RowSet &rows, VectorPtr *result) override {
            getIntValues(rows, requestedType_, result);
        }

    protected:
        // Switches based on filter type between different readHelper instantiations.
        template<
            typename Reader,
            bool isDense,
            bool kEncodingHasNulls,
            typename ExtractValues>
        void processFilter(
            pollux::common::Filter *filter,
            ExtractValues extractValues,
            const RowSet &rows);

        // Switches based on the type of ValueHook between different readWithVisitor
        // instantiations.
        template<typename Reader, bool isDense>
        void processValueHook(const RowSet &rows, ValueHook *hook);

        // Instantiates a Visitor based on type, isDense, value processing.
        template<
            typename Reader,
            typename TFilter,
            bool isDense,
            typename ExtractValues>
        void readHelper(
            pollux::common::Filter *filter,
            const RowSet &rows,
            ExtractValues extractValues);

        // The common part of integer reading. calls the appropriate
        // instantiation of processValueHook or processFilter based on
        // possible value hook, filter and denseness.
        template<typename Reader, bool kEncodingHasNulls>
        void readCommon(const RowSet &rows);
    };

    template<
        typename Reader,
        typename TFilter,
        bool isDense,
        typename ExtractValues>
    void SelectiveIntegerColumnReader::readHelper(
        pollux::common::Filter *filter,
        const RowSet &rows,
        ExtractValues extractValues) {
        switch (valueSize_) {
            case 2:
                reinterpret_cast<Reader *>(this)->Reader::readWithVisitor(
                    rows,
                    ColumnVisitor<int16_t, TFilter, ExtractValues, isDense>(
                        *reinterpret_cast<TFilter *>(filter), this, rows, extractValues));
                break;

            case 4:
                reinterpret_cast<Reader *>(this)->Reader::readWithVisitor(
                    rows,
                    ColumnVisitor<int32_t, TFilter, ExtractValues, isDense>(
                        *reinterpret_cast<TFilter *>(filter), this, rows, extractValues));
                break;

            case 8:
                reinterpret_cast<Reader *>(this)->Reader::readWithVisitor(
                    rows,
                    ColumnVisitor<int64_t, TFilter, ExtractValues, isDense>(
                        *reinterpret_cast<TFilter *>(filter), this, rows, extractValues));
                break;

            case 16:
                reinterpret_cast<Reader *>(this)->Reader::readWithVisitor(
                    rows,
                    ColumnVisitor<int128_t, TFilter, ExtractValues, isDense>(
                        *reinterpret_cast<TFilter *>(filter), this, rows, extractValues));
                break;

            default:
                POLLUX_FAIL("Unsupported valueSize_ {}", valueSize_);
        }
    }

    template<
        typename Reader,
        bool isDense,
        bool kEncodingHasNulls,
        typename ExtractValues>
    void SelectiveIntegerColumnReader::processFilter(
        pollux::common::Filter *filter,
        ExtractValues extractValues,
        const RowSet &rows) {
        if (filter == nullptr) {
            static_cast<Reader *>(this)
                    ->template readHelper<Reader, pollux::common::AlwaysTrue, isDense>(
                        &dwio::common::alwaysTrue(), rows, extractValues);
            return;
        }

        switch (filter->kind()) {
            case pollux::common::FilterKind::kAlwaysTrue:
                static_cast<Reader *>(this)
                        ->template readHelper<Reader, pollux::common::AlwaysTrue, isDense>(
                            filter, rows, extractValues);
                break;
            case pollux::common::FilterKind::kIsNull:
                if constexpr (kEncodingHasNulls) {
                    filterNulls<int64_t>(
                        rows, true, !std::is_same_v<decltype(extractValues), DropValues>);
                } else {
                    static_cast<Reader *>(this)
                            ->template readHelper<Reader, pollux::common::IsNull, isDense>(
                                filter, rows, extractValues);
                }
                break;
            case pollux::common::FilterKind::kIsNotNull:
                if constexpr (
                    kEncodingHasNulls &&
                    std::is_same_v<decltype(extractValues), DropValues>) {
                    filterNulls<int64_t>(rows, false, false);
                } else {
                    static_cast<Reader *>(this)
                            ->template readHelper<Reader, pollux::common::IsNotNull, isDense>(
                                filter, rows, extractValues);
                }
                break;
            case pollux::common::FilterKind::kBigintRange:
                static_cast<Reader *>(this)
                        ->template readHelper<Reader, pollux::common::BigintRange, isDense>(
                            filter, rows, extractValues);
                break;
            case pollux::common::FilterKind::kNegatedBigintRange:
                static_cast<Reader *>(this)
                        ->template readHelper<
                            Reader,
                            pollux::common::NegatedBigintRange,
                            isDense>(filter, rows, extractValues);
                break;
            case pollux::common::FilterKind::kBigintValuesUsingHashTable:
                static_cast<Reader *>(this)
                        ->template readHelper<
                            Reader,
                            pollux::common::BigintValuesUsingHashTable,
                            isDense>(filter, rows, extractValues);
                break;
            case pollux::common::FilterKind::kBigintValuesUsingBitmask:
                static_cast<Reader *>(this)
                        ->template readHelper<
                            Reader,
                            pollux::common::BigintValuesUsingBitmask,
                            isDense>(filter, rows, extractValues);
                break;
            case pollux::common::FilterKind::kNegatedBigintValuesUsingHashTable:
                static_cast<Reader *>(this)
                        ->template readHelper<
                            Reader,
                            pollux::common::NegatedBigintValuesUsingHashTable,
                            isDense>(filter, rows, extractValues);
                break;
            case pollux::common::FilterKind::kNegatedBigintValuesUsingBitmask:
                static_cast<Reader *>(this)
                        ->template readHelper<
                            Reader,
                            pollux::common::NegatedBigintValuesUsingBitmask,
                            isDense>(filter, rows, extractValues);
                break;
            default:
                static_cast<Reader *>(this)
                        ->template readHelper<Reader, pollux::common::Filter, isDense>(
                            filter, rows, extractValues);
                break;
        }
    }

    template<typename Reader, bool isDense>
    void SelectiveIntegerColumnReader::processValueHook(
        const RowSet &rows,
        ValueHook *hook) {
        switch (hook->kind()) {
            case aggregate::AggregationHook::kBigintSum:
                readHelper<Reader, pollux::common::AlwaysTrue, isDense>(
                    &alwaysTrue(),
                    rows,
                    ExtractToHook<aggregate::SumHook<int64_t, false> >(hook));
                break;
            case aggregate::AggregationHook::kBigintSumOverflow:
                readHelper<Reader, pollux::common::AlwaysTrue, isDense>(
                    &alwaysTrue(),
                    rows,
                    ExtractToHook<aggregate::SumHook<int64_t, true> >(hook));
                break;
            case aggregate::AggregationHook::kBigintMax:
                readHelper<Reader, pollux::common::AlwaysTrue, isDense>(
                    &dwio::common::alwaysTrue(),
                    rows,
                    ExtractToHook<aggregate::MinMaxHook<int64_t, false> >(hook));
                break;
            case aggregate::AggregationHook::kBigintMin:
                readHelper<Reader, pollux::common::AlwaysTrue, isDense>(
                    &alwaysTrue(),
                    rows,
                    ExtractToHook<aggregate::MinMaxHook<int64_t, true> >(hook));
                break;
            default:
                readHelper<Reader, pollux::common::AlwaysTrue, isDense>(
                    &alwaysTrue(), rows, ExtractToGenericHook(hook));
        }
    }

    template<typename Reader, bool kEncodingHasNulls>
    void SelectiveIntegerColumnReader::readCommon(const RowSet &rows) {
        const bool isDense = rows.back() == rows.size() - 1;
        pollux::common::Filter *filter =
                scanSpec_->filter() ? scanSpec_->filter() : &alwaysTrue();
        if (scanSpec_->keepValues()) {
            if (scanSpec_->valueHook()) {
                if (isDense) {
                    processValueHook<Reader, true>(rows, scanSpec_->valueHook());
                } else {
                    processValueHook<Reader, false>(rows, scanSpec_->valueHook());
                }
            } else {
                if (isDense) {
                    processFilter<Reader, true, kEncodingHasNulls>(
                        filter, ExtractToReader(this), rows);
                } else {
                    processFilter<Reader, false, kEncodingHasNulls>(
                        filter, ExtractToReader(this), rows);
                }
            }
        } else {
            if (isDense) {
                processFilter<Reader, true, kEncodingHasNulls>(
                    filter, DropValues(), rows);
            } else {
                processFilter<Reader, false, kEncodingHasNulls>(
                    filter, DropValues(), rows);
            }
        }
    }
} // namespace kumo::pollux::dwio::common
